# -*- coding: utf-8 -*-
from airflow.providers.dingding.operators.dingding import DingdingHook
from airflow.providers.presto.hooks.presto import PrestoHook

from utils.macros import cst_ds_nodash


def ml_redis_check(at_mobiles=None, at_all=None):
    def _ml_redis_check(ctx):
        execution_date = cst_ds_nodash(ctx["execution_date"])
        # dingding_ThreeSeg_etl_info、dingding_ThreeSeg_etl_alert
        dingding_conn_id = "dingding_redis_ml"
        # dingding_conn_id = "dingding_bigdata"
        ph = PrestoHook(presto_conn_id="presto_default")
        env = ctx['var']['value'].env
        sql1 = f"""with t_second as (
    select network_code, code, dt, area_id
    from dim.yl_cn_mysql_lmdm_sys_second_code
    where dt = '{execution_date}'
      and is_enable = 1
      and is_delete = 1
    group by network_code, code, dt, area_id
),
     t_area as (
         select destination_area_id, code as fist_code
         from (select code, destination_area_id,row_number() over (partition by destination_area_id) rnk
               from dim.yl_cn_mysql_lmdm_sys_first_code
               where dt = '{execution_date}'
                 and is_enable = 1
                 and is_delete = 1) t
         where t.rnk=1
     )
select count(1)
from (
         select t_second.network_code,
                t_second.code,
                t_second.dt,
                concat(if(t_second.area_id is null, '', cast(t_second.area_id as varchar )), '-',
                       if(t_area.fist_code is null, '', t_area.fist_code)) as first_code,
                concat(cast(t_second.area_id as varchar ), '-', t_area.fist_code)            as flag
         from t_second t_second
                  left join t_area t_area
                            on t_second.area_id = t_area.destination_area_id
     ) a
where a.flag is null"""

        sql2 = f"""with t_second as (
    select network_code, code, dt, area_id
    from dim.yl_cn_mysql_lmdm_sys_second_code
    where dt = '{execution_date}'
      and is_enable = 1
      and is_delete = 1
    group by network_code, code, dt, area_id
),
     t_area as (
         select destination_area_id, code as fist_code
         from (select code, destination_area_id,row_number() over (partition by destination_area_id) rnk
               from dim.yl_cn_mysql_lmdm_sys_first_code
               where dt = '{execution_date}'
                 and is_enable = 1
                 and is_delete = 1) t
         where t.rnk=1
     )
select network_code, first_code
from (
         select t_second.network_code,
                t_second.code,
                t_second.dt,
                concat(if(t_second.area_id is null, '', cast(t_second.area_id as varchar )), '-',
                       if(t_area.fist_code is null, '', t_area.fist_code)) as first_code,
                concat(cast(t_second.area_id as varchar ), '-', t_area.fist_code)            as flag
         from t_second t_second
                  left join t_area t_area
                            on t_second.area_id = t_area.destination_area_id
     ) a
where a.flag is null"""

        cnt1 = ph.get_records(sql1)[0]
        cnt2 = ph.get_records(sql2)
        title = f"【{env}】redis一段码缺失监控"
        text = f"""
# {title}
   - *分区* ：{execution_date}
   - *缺失数据量* ：{cnt1}
   - *[网点编码,区域id]* ：{cnt2}
"""
        if cnt1[0] > 0:
            return DingdingHook(dingding_conn_id=dingding_conn_id,
                                message_type='markdown',  # text, link, markdown, actionCard, feedCard
                                message={'title': title, 'text': text},
                                at_mobiles=at_mobiles,
                                at_all=at_all, ).send()

    return _ml_redis_check
